package com.navinfo.tripanalysis.offline.service.impl;

import com.navinfo.tripanalysis.common.pojo.Point;
import com.navinfo.tripanalysis.common.pojo.Point0F37;
import com.navinfo.tripanalysis.offline.pojo.*;
import com.navinfo.tripanalysis.offline.util.ExcelUtils;
import com.navinfo.tripanalysis.common.util.DateUtils;
import com.navinfo.tripanalysis.offline.service.LoadPoint0F37Service;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;

import java.io.File;
import java.util.ArrayList;
import java.util.List;

/**
 * excel中加载0f37数据
 * @author wangshuai
 */
public class LoadPoint0F37ServiceExcelImpl implements LoadPoint0F37Service {
    private String path = "";
    public String getPath() {
        return path;
    }
    public void setPath(String path) {
        this.path = path;
    }

    @Override
    public JavaRDD<Point> load(SparkSession spark, JavaSparkContext jsc, LoadDataParam param) {
        String dayStr = DateUtils.format(param.getDay(), DateUtils.DateFormat.YYYYMMDD);
        String type = "0f37";
        File file = new File(getPath() + File.separator + type+ File.separator + dayStr + File.separator );
        try {
            List<Point> list = new ArrayList<>(10000);
            if (file.exists()) {
                File[] files = file.listFiles();
                for (File f : files) {
                    String name = f.getName();
                    if (StringUtils.endsWith(name, "xlsx")) {
                        List<Point> sublist = (List) ExcelUtils.readExcel(f, Point0F37.class);
                        list.addAll(sublist);
                    }
                }
            }
            return jsc.parallelize(list);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return null;
    }
}
